package org.ligson.hbase127.reposiotry;

import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.commons.beanutils.BeanUtilsBean2;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.reflect.FieldUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.ligson.hbase127.core.HBaseEntityInformation;
import org.ligson.hbase127.fw.Family;
import org.ligson.hbase127.fw.IdGenerator;
import org.ligson.hbase127.fw.RowIdGenerator;
import org.ligson.hbase127.fw.Table;
import org.springframework.beans.BeanUtils;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

@Repository
@Transactional(readOnly = true)
public class SimpleHBaseRepository<T, ID> implements HBaseRepository<T, ID> {
    private HBaseEntityInformation hBaseEntityInformation;
    private HbaseTemplate hbaseTemplate;

    public SimpleHBaseRepository(HBaseEntityInformation hBaseEntityInformation,
                                 HbaseTemplate hbaseTemplatey) {
        this.hBaseEntityInformation = hBaseEntityInformation;
        this.hbaseTemplate = hbaseTemplate;
    }

    public SimpleHBaseRepository(HBaseEntityInformation hBaseEntityInformation) {
        this.hBaseEntityInformation = hBaseEntityInformation;
    }

    public SimpleHBaseRepository() {
    }

    public static byte[] convertFieldToBytes(Object instance, String fieldName) {
        Class<?> domainType = instance.getClass();
        Field field = FieldUtils.getDeclaredField(instance.getClass(), fieldName);
        Object value;
        try {
            value = field.get(instance);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
        Method method = BeanUtils.findMethod(Bytes.class, "toBytes");
        if (method == null) {
            throw new RuntimeException(String.format("实体:%s字段:%s不支持的类型:%s转换", domainType.getSimpleName(), field.getName(), field.getType().getSimpleName()));
        }
        byte[] buffer;
        try {
            buffer = (byte[]) method.invoke(null, value);
        } catch (Exception e) {
            throw new RuntimeException(String.format("实体:%s字段:%s的类型:%s不能转换为字节", domainType.getSimpleName(), field.getName(), field.getType().getSimpleName()));
        }
        return buffer;
    }

    private byte[] convertObj2Bytes(Object object) {
        Method method = BeanUtils.findMethod(Bytes.class, "toBytes");
        byte[] buffer;
        try {
            buffer = (byte[]) method.invoke(null, object);
        } catch (Exception e) {
            throw new RuntimeException(String.format("对象:%s类型:%s不能转bytes", object, object.getClass().getSimpleName()));
        }
        return buffer;
    }

    @Override
    public <S extends T> S save(S entity) {
        Class domainType = hBaseEntityInformation.getJavaType();
        Table table = (Table) domainType.getDeclaredAnnotation(Table.class);
        PropertyDescriptor[] pds = BeanUtils.getPropertyDescriptors(domainType);
        Class<? extends RowIdGenerator> rowIdGeneratorClazz = table.rowIdGenerator();
        RowIdGenerator rowIdGenerator;
        try {
            rowIdGenerator = rowIdGeneratorClazz.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(String.format("实例化实体：%s的RowIdGenerator:%s失败", domainType.getName(), rowIdGeneratorClazz.getName()));
        }
        String rowId = rowIdGenerator.generator(entity);
        boolean isNew = hBaseEntityInformation.isNew(entity);
        for (PropertyDescriptor pd : pds) {
            Field field = FieldUtils.getDeclaredField(domainType, pd.getName());
            Family family = field.getDeclaredAnnotation(Family.class);
            byte[] buffer;
            String fieldName;
            String id = null;

            if (isNew && field.getName().equals(table.idName())) {
                Class<? extends IdGenerator> idge = table.idGenerator();
                IdGenerator idGenerator;
                try {
                    idGenerator = idge.newInstance();
                } catch (Exception e) {
                    throw new RuntimeException(String.format("实例化实体%s的主键生成策略%s失败", domainType.getSimpleName(), idge.getSimpleName()));
                }
                id = idGenerator.generator(entity);
                buffer = Bytes.toBytes(id);
                fieldName = table.idName();
            } else {
                org.ligson.hbase127.fw.Field fields = field.getDeclaredAnnotation(org.ligson.hbase127.fw.Field.class);
                if (family == null || fields == null) {
                    continue;
                }
                buffer = convertFieldToBytes(entity, field.getName());
                fieldName = fields.value();
            }
            if (StringUtils.isBlank(fieldName)) {
                fieldName = field.getName();
            }
            hbaseTemplate.put(table.value(), rowId, family.value(), fieldName, buffer);
            if (id != null) {
                try {
                    FieldUtils.writeDeclaredField(entity, field.getName(), id);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return entity;
    }

    @Override
    public <S extends T> Iterable<S> saveAll(Iterable<S> entities) {
        List<S> list = new ArrayList<>();
        entities.forEach(e -> list.add(save(e)));
        return list;
    }


    @Override
    public Optional<T> findById(ID id) {


        Class<T> domainType = hBaseEntityInformation.getJavaType();
        Table table = (Table) domainType.getDeclaredAnnotation(Table.class);
        Field idField = FieldUtils.getDeclaredField(domainType, table.idName());
        Family family = idField.getDeclaredAnnotation(Family.class);

        List<Filter> filters = new ArrayList<>();
        Filter filter0 = new SingleColumnValueFilter(Bytes.toBytes(family.value()), Bytes.toBytes(idField.getName()), CompareFilter.CompareOp.EQUAL, convertObj2Bytes(id));
        filters.add(filter0);
        FilterList filterList1 = new FilterList(filters);

        Scan scan = new Scan();
        scan.setFilter(filterList1);

        List<T> list = hbaseTemplate.find(table.value(), scan, new RowMapper<T>() {
            @Override
            public T mapRow(Result result, int rowNum) throws Exception {
                return convertResultToEntity(domainType, result, rowNum);
            }
        });
        return Optional.ofNullable(CollectionUtils.isEmpty(list) ? null : list.get(0));
    }

    @Override
    public boolean existsById(Object o) {
        Object id = hBaseEntityInformation.getId(o);
        return findById((ID) id).isPresent();
    }

    private T convertResultToEntity(Class<T> domainType, Result result, int rowNum) throws Exception {
        Object obj = domainType.newInstance();
        List<Cell> listCells = result.listCells();
        BeanUtilsBean instance = BeanUtilsBean2.getInstance();
        for (Cell cell : listCells) {
            byte[] familyArray = CellUtil.cloneFamily(cell);
            byte[] qualifierArray = CellUtil.cloneQualifier(cell);
            byte[] valueArray = CellUtil.cloneValue(cell);
            String fieldName = Bytes.toString(qualifierArray);
            PropertyDescriptor pd = BeanUtils.getPropertyDescriptor(domainType, fieldName);
            if (pd == null) {
                throw new RuntimeException(String.format("实体:%s字段:%s不存在", domainType.getSimpleName(), fieldName));
            }
            Class<?> pdType = pd.getPropertyType();
            Method method = BeanUtils.findMethod(Bytes.class, "to" + pdType.getSimpleName(), (new byte[]{}).getClass());
            if (method == null) {
                throw new RuntimeException(String.format("实体:%s字段:%s不支持的类型:%s转换", domainType.getSimpleName(), fieldName, pdType.getSimpleName()));
            }
//                    pd.setValue(fieldName, method.invoke(null, (Object) valueArray));
            instance.setProperty(obj, fieldName, method.invoke(null, (Object) valueArray));
        }
        return (T) obj;
    }

    @Override
    public List<T> findAll() {
        Class<T> domainType = hBaseEntityInformation.getJavaType();
        Table table = domainType.getDeclaredAnnotation(Table.class);
        return hbaseTemplate.find(table.value(), new Scan(), new RowMapper<T>() {
            @Override
            public T mapRow(Result result, int rowNum) throws Exception {
                return convertResultToEntity(domainType, result, rowNum);
            }
        });
    }

    @Override
    public Iterable<T> findAllById(Iterable<ID> ids) {
        List<T> list = new ArrayList<>();
        for (ID id : ids) {
            Optional<T> e = findById(id);
            e.ifPresent(list::add);
        }
        return list;
    }

    @Override
    public long count() {
        return findAll().size();
    }

    @Override
    public void deleteById(Object o) {
        Class domainType = hBaseEntityInformation.getJavaType();
        Table table = (Table) domainType.getDeclaredAnnotation(Table.class);
        //PropertyDescriptor[] pds = BeanUtils.getPropertyDescriptors(domainType);
        Class<? extends RowIdGenerator> rowIdGeneratorClazz = table.rowIdGenerator();
        RowIdGenerator rowIdGenerator;
        try {
            rowIdGenerator = rowIdGeneratorClazz.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(String.format("实例化实体：%s的RowIdGenerator:%s失败", domainType.getName(), rowIdGeneratorClazz.getName()));
        }
        String rowId = rowIdGenerator.generator(o);
        Field idField = FieldUtils.getDeclaredField(domainType, table.idName());
        Family family = idField.getDeclaredAnnotation(Family.class);
        hbaseTemplate.delete(table.value(), rowId, family.value(), table.idName());
    }

    @Override
    public void delete(Object entity) {
        Class<T> domainType = hBaseEntityInformation.getJavaType();
        Table table = (Table) domainType.getDeclaredAnnotation(Table.class);
        Field idField = FieldUtils.getDeclaredField(domainType, table.idName());
        Family family = idField.getDeclaredAnnotation(Family.class);
        PropertyDescriptor[] pds = BeanUtils.getPropertyDescriptors(domainType);

        Scan scan = new Scan();
        List<Filter> filters = new ArrayList<>();
        for (PropertyDescriptor pd : pds) {
            Object obj = pd.getValue(pd.getName());
            byte[] values = convertObj2Bytes(obj);
            Filter filter0 = new SingleColumnValueFilter(Bytes.toBytes(family.value()), Bytes.toBytes(idField.getName()), CompareFilter.CompareOp.EQUAL, convertObj2Bytes(values));
            filters.add(filter0);
            FilterList filterList1 = new FilterList(filters);
            scan.setFilter(filterList1);
        }
        List<T> list = hbaseTemplate.find(table.value(), scan, new RowMapper<T>() {
            @Override
            public T mapRow(Result result, int rowNum) throws Exception {
                return convertResultToEntity(domainType, result, rowNum);
            }
        });
        for (T t : list) {
            Object id = hBaseEntityInformation.getId(t);
            delete(id);
        }
    }

    @Override
    public void deleteAll(Iterable entities) {
        entities.forEach(this::delete);
    }

    @Override
    public void deleteAll() {
        deleteAll(findAll());
    }
}
